package io.kiki.stack.netty;


import io.kiki.stack.netty.channelManager.*;
import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.channelManager.ChannelInboundHandlerAdapter;
import io.kiki.stack.netty.command.CommandCode;
import io.kiki.stack.netty.command.CommandFactoryImpl;
import io.kiki.stack.netty.config.ConfigManager;
import io.kiki.stack.netty.config.ExtendedChannelHandlerList;
import io.kiki.stack.netty.config.GenericOption;
import io.kiki.stack.netty.config.ServerOption;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.invoker.*;
import io.kiki.stack.netty.processor.Processor;
import io.kiki.stack.netty.protocol.Codec;
import io.kiki.stack.netty.protocol.CodecImpl;
import io.kiki.stack.netty.protocol.ProtocolImpl;
import io.kiki.stack.netty.userProcessor.UserProcessor;
import io.kiki.stack.netty.userProcessor.UserProcessorRegisterHelper;
import io.kiki.stack.netty.util.*;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.security.KeyStore;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * Server for Rpc.
 * <p>
 * Usage:
 * You can initialize RpcServer with one of the three constructors:
 * {@link #Server(int)}, {@link #Server(int, boolean)}, {@link #Server(int, boolean, boolean)}
 * Then call start() to start a rpc server, and call stop() to stop a rpc server.
 * <p>
 * Notice:
 * Once rpc server has been stopped, it can never be start again. You should init another instance of RpcServer to use.
 */

@Slf4j
@Setter
@Getter
public class Server extends AbstractServer {
    /**
     * worker event loop group. Reuse I/O worker threads between rpc servers.
     */
    private static final EventLoopGroup EVENT_LOOP_GROUP = EventLoopGroupUtil.getEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2, new NamedThreadFactory("netty-server-worker", true));

    static {
        if (EVENT_LOOP_GROUP instanceof NioEventLoopGroup) {
            ((NioEventLoopGroup) EVENT_LOOP_GROUP).setIoRatio(ConfigManager.netty_io_ratio());
        } else
            if (EVENT_LOOP_GROUP instanceof EpollEventLoopGroup) {
                ((EpollEventLoopGroup) EVENT_LOOP_GROUP).setIoRatio(ConfigManager.netty_io_ratio());
            }
    }

    /**
     * boss event loop group, boss group should not be daemon, need shutdown manually
     */
    private final EventLoopGroup bossGroup = EventLoopGroupUtil.getEventLoopGroup(1, new NamedThreadFactory("Rpc-netty-server-boss", false));
    private final ChannelEventListener channelEventListener = new ChannelEventListener();
    /**
     * user processors of rpc server
     */
    private final ConcurrentHashMap<String, UserProcessor<?>> userProcessors = new ConcurrentHashMap<String, UserProcessor<?>>(4);

    private final Codec codec = new CodecImpl();

    protected Invoker invoker;

    private ServerBootstrap serverBootstrap;

    private ChannelFuture channelFuture;

    private ChannelEventHandler channelEventHandler;


    private ServerChannelManagerImpl serverChannelManager;

    /**
     * Construct a rpc server with random port <br>
     * the random port will determined after server startup
     * Note:<br>
     * You can only use invoke methods with params {@link Channel}, for example {@link #invokeSync(Channel, Object, int)} <br>
     * Otherwise {@link UnsupportedOperationException} will be thrown.
     */
    public Server() {
        this(false);
    }

    /**
     * Construct a rpc server with random port <br>
     * the random port will determined after server startup
     * Note:<br>
     * You can only use invoke methods with params {@link Channel}, for example {@link #invokeSync(Channel, Object, int)} <br>
     * Otherwise {@link UnsupportedOperationException} will be thrown.
     *
     * @param manageChannel true to enable channel management feature
     */
    public Server(boolean manageChannel) {
        this(0, manageChannel);
    }

    /**
     * Construct a rpc server. <br>
     * <p>
     * Note:<br>
     * You can only use invoke methods with params {@link Channel}, for example {@link #invokeSync(Channel, Object, int)} <br>
     * Otherwise {@link UnsupportedOperationException} will be thrown.
     */
    public Server(int port) {
        this(port, false);
    }

    /**
     * Construct a rpc server. <br>
     * <p>
     * Note:<br>
     * You can only use invoke methods with params {@link Channel}, for example {@link #invokeSync(Channel, Object, int)} <br>
     * Otherwise {@link UnsupportedOperationException} will be thrown.
     */
    public Server(String ip, int port) {
        this(ip, port, false);
    }

    /**
     * Construct a rpc server. <br>
     *
     * <ul>
     * <li>You can enable channel management feature by specify @param managechannel true.</li>
     * <ul>
     * <li>When channel management feature enabled, you can use all invoke methods with params {@link String}, {@link Url}, {@link Channel} methods.</li>
     * <li>When channel management feature disabled, you can only use invoke methods with params {@link Channel}, otherwise {@link UnsupportedOperationException} will be thrown.</li>
     * </ul>
     * </ul>
     *
     * @param port          listened port
     * @param manageChannel true to enable channel management feature
     */
    public Server(int port, boolean manageChannel) {
        super(port);
        /* server channel management feature enabled or not, default value false, means disabled. */
        if (manageChannel) {
            option(ServerOption.SERVER_MANAGE_channel_SWITCH, true);
        }
    }

    /**
     * Construct a rpc server. <br>
     *
     * <ul>
     * <li>You can enable channel management feature by specify @param managechannel true.</li>
     * <ul>
     * <li>When channel management feature enabled, you can use all invoke methods with params {@link String}, {@link Url}, {@link Channel} methods.</li>
     * <li>When channel management feature disabled, you can only use invoke methods with params {@link Channel}, otherwise {@link UnsupportedOperationException} will be thrown.</li>
     * </ul>
     * </ul>
     *
     * @param port          listened port
     * @param managechannel true to enable channel management feature
     */
    public Server(String ip, int port, boolean managechannel) {
        super(ip, port);
        /* server channel management feature enabled or not, default value false, means disabled. */
        if (managechannel) {
            option(ServerOption.SERVER_MANAGE_channel_SWITCH, true);
        }
    }

    /**
     * Construct a rpc server. <br>
     * <p>
     * You can construct a rpc server with synchronous or asynchronous stop strategy by {@param syncStop}.
     *
     * @param port          listened port
     * @param managechannel manage channel
     * @param syncStop      true to enable stop in synchronous way
     */
    public Server(int port, boolean managechannel, boolean syncStop) {
        this(port, managechannel);
        if (syncStop) {
            option(ServerOption.SERVER_SYNC_STOP, true);
        }
    }

    @Override
    protected void doInit() {

        if (option(ServerOption.SERVER_MANAGE_channel_SWITCH)) {
            // in server side, do not care the channel service state, so use null instead of global switch
            ChannelSelectStrategy channelSelectStrategy = new RandomChannelSelectStrategy(this);
            this.serverChannelManager = new ServerChannelManagerImpl(channelSelectStrategy);
            this.serverChannelManager.startup();

            this.channelEventHandler = new RpcChannelEventHandler(this);
            this.channelEventHandler.setChannelManager(this.serverChannelManager);
            this.channelEventHandler.setChannelEventListener(this.channelEventListener);
        } else {
            this.channelEventHandler = new ChannelEventHandler(this);
            this.channelEventHandler.setChannelEventListener(this.channelEventListener);
        }
        initRpcRemoting();

        Integer tcpSoSndBuf = option(GenericOption.tcp_so_sndbuf);
        Integer tcpSoRcvBuf = option(GenericOption.tcp_so_rcvbuf);

        this.serverBootstrap = new ServerBootstrap();
        this.serverBootstrap.group(bossGroup, EVENT_LOOP_GROUP).channel(EventLoopGroupUtil.getServerSocketChannelClass()).option(ChannelOption.SO_BACKLOG, ConfigManager.tcp_so_backlog()).option(ChannelOption.SO_REUSEADDR, ConfigManager.tcp_so_reuseaddr()).childOption(ChannelOption.TCP_NODELAY, ConfigManager.tcp_nodelay()).childOption(ChannelOption.SO_KEEPALIVE, ConfigManager.tcp_so_keepalive()).childOption(ChannelOption.SO_SNDBUF, tcpSoSndBuf != null ? tcpSoSndBuf : ConfigManager.tcp_so_sndbuf()).childOption(ChannelOption.SO_RCVBUF, tcpSoRcvBuf != null ? tcpSoRcvBuf : ConfigManager.tcp_so_rcvbuf());

        // set write buffer water mark
        initWriteBufferWaterMark();

        // init byte buf allocator
        if (ConfigManager.netty_buffer_pooled()) {
            this.serverBootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT).childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        } else {
            this.serverBootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT).childOption(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
        }

        // enable trigger mode for epoll if need
        EventLoopGroupUtil.enableEpollMode(serverBootstrap);

        final boolean idleSwitch = ConfigManager.tcp_idle_switch();
        final boolean flushConsolidationSwitch = option(ServerOption.netty_flush_consolidation);
        final int idleTime = ConfigManager.tcp_server_idle();
        final ChannelHandler serverIdleHandler = new ServerIdleHandler();
        final ChannelInboundHandlerAdapter channelInboundHandlerAdapter = new ChannelInboundHandlerAdapter(true, this.userProcessors);
        this.serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel channel) {
                ChannelPipeline pipeline = channel.pipeline();
                ExtendedChannelHandlerList extendedHandlers = option(ServerOption.extended_netty_channel_handler);
                if (extendedHandlers != null) {
                    List<ChannelHandler> frontHandlers = extendedHandlers.frontChannelHandlers();
                    if (frontHandlers != null) {
                        for (ChannelHandler channelHandler : frontHandlers) {
                            pipeline.addLast(channelHandler.getClass().getName(), channelHandler);
                        }
                    }
                }

                Boolean sslEnable = option(ServerOption.SRV_SSL_ENABLE);
                if (!sslEnable) {
                    // fixme: remove in next version
                    sslEnable = RpcConfigManager.server_ssl_enable();
                }
                if (sslEnable) {
                    SSLEngine sslEngine = initSSLContext().newEngine(channel.alloc());
                    sslEngine.setUseClientMode(false);
                    // fixme: update in next version
                    sslEngine.setNeedClientAuth(option(ServerOption.SRV_SSL_NEED_CLIENT_AUTH) || RpcConfigManager.server_ssl_need_client_auth());
                    pipeline.addLast(Constants.ssl_handler, new SslHandler(sslEngine));
                }

                if (flushConsolidationSwitch) {
                    pipeline.addLast("flushConsolidationHandler", new FlushConsolidationHandler(1024, true));
                }
                pipeline.addLast("decoder", codec.decoder());
                pipeline.addLast("encoder", codec.encoder());
                if (idleSwitch) {
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(0, 0, idleTime, TimeUnit.MILLISECONDS));
                    pipeline.addLast("serverIdleHandler", serverIdleHandler);
                }
                pipeline.addLast("channelEventHandler", channelEventHandler);
                pipeline.addLast("handler", channelInboundHandlerAdapter);
                if (extendedHandlers != null) {
                    List<ChannelHandler> backHandlers = extendedHandlers.backChannelHandlers();
                    if (backHandlers != null) {
                        for (ChannelHandler channelHandler : backHandlers) {
                            pipeline.addLast(channelHandler.getClass().getName(), channelHandler);
                        }
                    }
                }
                createchannel(channel);
            }

            /**
             * create channel operation<br>
             * <ul>
             * <li>If flag managechannel be true, use {@link ChannelManagerImpl} to add a new channel, meanwhile bind it with the channel.</li>
             * <li>If flag managechannel be false, just create a new channel and bind it with the channel.</li>
             * </ul>
             */
            private void createchannel(SocketChannel channel) {
                Url url = AddressParser.parse(InetSocketAddressUtil.parseRemoteAddress(channel));
                if (option(ServerOption.SERVER_MANAGE_channel_SWITCH)) {
                    serverChannelManager.add(new Channel(channel, url), url.getId());
                } else {
                    new Channel(channel, url);
                }
                channel.pipeline().fireUserEventTriggered(ChannelEventType.CONNECT);
            }
        });
    }

    private SslContext initSSLContext() {
        InputStream inputStream = null;
        try {
            String keyStoreType = option(ServerOption.SRV_SSL_KEYSTORE_TYPE);
            if (keyStoreType == null) {
                keyStoreType = RpcConfigManager.server_ssl_keystore_type();
            }
            KeyStore keyStore = KeyStore.getInstance(keyStoreType);
            String filePath = option(ServerOption.SRV_SSL_KEYSTORE);
            if (filePath == null) {
                filePath = RpcConfigManager.server_ssl_keystore();
            }
            inputStream = new FileInputStream(filePath);

            String keyStorePass = option(ServerOption.SRV_SSL_KEYSTORE_PASS);
            if (keyStorePass == null) {
                keyStorePass = RpcConfigManager.server_ssl_keystore_pass();
            }
            char[] passChs = keyStorePass.toCharArray();
            keyStore.load(inputStream, passChs);
            String sslAlgorithm = RpcConfigManager.server_ssl_kmf_algorithm();
            if (sslAlgorithm == null) {
                sslAlgorithm = option(ServerOption.SRV_SSL_KMF_ALGO);
            }
            KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(sslAlgorithm);
            keyManagerFactory.init(keyStore, passChs);

            TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(RpcConfigManager.client_ssl_tmf_algorithm());
            trustManagerFactory.init(keyStore);
            return SslContextBuilder.forServer(keyManagerFactory).trustManager(trustManagerFactory).build();
        } catch (Exception e) {
            //log.error("Fail to init SSL context for server.", e);
            throw new IllegalStateException("Fail to init SSL context", e);
        } finally {
            IoUtils.closeQuietly(inputStream);
        }

    }

    @Override
    protected boolean doStart() throws InterruptedException {
        for (UserProcessor<?> userProcessor : userProcessors.values()) {
            if (!userProcessor.isInited()) {
                userProcessor.startup();
            }
        }

        this.channelFuture = this.serverBootstrap.bind(new InetSocketAddress(ip(), port())).sync();
        if (port() == 0 && channelFuture.isSuccess()) {
            InetSocketAddress localAddress = (InetSocketAddress) channelFuture.channel().localAddress();
            setLocalBindingPort(localAddress.getPort());
            //log.info("rpc server start with random port: {}!", port());
        }
        return this.channelFuture.isSuccess();
    }

    /**
     * Notice: only RpcServer#option(ServerOption.SERVER_MANAGE_channel_SWITCH, true) switch on, will close all channels.
     *
     * @see AbstractServer#doStop()
     */
    @Override
    protected boolean doStop() {
        if (null != this.channelFuture) {
            this.channelFuture.channel().close();
        }
        if (option(ServerOption.SERVER_SYNC_STOP)) {
            this.bossGroup.shutdownGracefully().awaitUninterruptibly();
        } else {
            this.bossGroup.shutdownGracefully();
        }
        if (option(ServerOption.SERVER_MANAGE_channel_SWITCH) && null != this.serverChannelManager) {
            this.serverChannelManager.shutdown();
            //log.warn("Close all channels from server side!");
        }
        for (UserProcessor<?> userProcessor : userProcessors.values()) {
            if (userProcessor.isInited()) {
                userProcessor.shutdown();
            }
        }
        //log.warn("Rpc Server stopped!");
        return true;
    }

    /**
     * init rpc remoting
     */
    protected void initRpcRemoting() {
        this.invoker = new ServerInvoker(new CommandFactoryImpl(), this.serverChannelManager);
    }

    /**
     * @see IServer#registerProcessor(byte, CommandCode, Processor)
     */
    @Override
    public void registerProcessor(byte protocolCode, CommandCode cmd, Processor<?> processor) {
        ProtocolImpl.protocol.getCommandHandler().registerProcessor(cmd, processor);
    }

    /**
     * @see IServer#registerDefaultExecutor(byte, Executor)
     */
    @Override
    public void registerDefaultExecutor(byte protocolCode, Executor executor) {
        ProtocolImpl.protocol.getCommandHandler().registerDefaultExecutor(executor);
    }

    /**
     * Add processor to process channel event.
     *
     * @param type      channel event type
     * @param processor channel event processor
     */
    public void addChannelEventProcessor(ChannelEventType type, ChannelEventProcessor processor) {
        this.channelEventListener.add(type, processor);
    }

    /**
     * Use UserProcessorRegisterHelper{@link UserProcessorRegisterHelper} to help register user processor for server side.
     *
     * @see AbstractServer#registerUserProcessor(UserProcessor)
     */
    @Override
    public void registerUserProcessor(UserProcessor<?> userProcessor) {
        UserProcessorRegisterHelper.registerUserProcessor(userProcessor, this.userProcessors);
        // startup the processor if it registered after component startup
        if (isInited() && !userProcessor.isInited()) {
            userProcessor.startup();
        }
    }

    /**
     * One way invocation using a string address, address format example - 127.0.0.1:12200?key1=value1&key2=value2 <br>
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the string address to find a available client channel, if none then throw exception</li>
     * <li>Unlike rpc client, address arguments takes no effect here, for rpc server will not create channel.</li>
     * </ol>
     *
     * @param addr    address addr
     * @param request request request
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void oneway(final String addr, final Object request) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.oneway(addr, request, null);
    }

    /**
     * One way invocation with a {@link RequestContext}, common api notice please see {@link #oneway(String, Object)}
     *
     * @param addr           address
     * @param request        request
     * @param requestContext invokeContext
     * @throws RemotingException    remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void oneway(final String addr, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.oneway(addr, request, requestContext);
    }

    /**
     * One way invocation using a parsed {@link Url} <br>
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the parsed {@link Url} to find a available client channel, if none then throw exception</li>
     * </ol>
     *
     * @param url     url
     * @param request request request
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void oneway(final Url url, final Object request) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.oneway(url, request, null);
    }

    /**
     * One way invocation with a {@link RequestContext}, common api notice please see {@link #oneway(Url, Object)}
     *
     * @param url            url
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void oneway(final Url url, final Object request, final RequestContext requestContext) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.oneway(url, request, requestContext);
    }

    /**
     * One way invocation using a {@link Channel} <br>
     * <p>
     * Notice:<br>
     * <b>DO NOT modify the request object concurrently when this method is called.</b>
     *
     * @param channel channel channel
     * @param request request request
     * @throws RemotingException remoting exception remoting exception
     */
    public void oneway(final Channel channel, final Object request) throws RemotingException {
        ensureStarted();
        this.invoker.oneway(channel, request, null);
    }

    /**
     * One way invocation with a {@link RequestContext}, common api notice please see {@link #oneway(Channel, Object)}
     *
     * @param channel        channel channel
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @throws RemotingException remoting exception remoting exception
     */
    public void oneway(final Channel channel, final Object request, final RequestContext requestContext) throws RemotingException {
        ensureStarted();
        this.invoker.oneway(channel, request, requestContext);
    }

    /**
     * Synchronous invocation using a string address, address format example - 127.0.0.1:12200?key1=value1&key2=value2 <br>
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the string address to find a available client channel, if none then throw exception</li>
     * <li>Unlike rpc client, address arguments takes no effect here, for rpc server will not create channel.</li>
     * </ol>
     *
     * @param addr          address addr
     * @param request       request request
     * @param timeoutMillis timeoutMillis timeoutMillis
     * @return Object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(final String addr, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeSync(addr, request, null, timeoutMillis);
    }

    /**
     * Synchronous invocation with a {@link RequestContext}, common api notice please see {@link #invokeSync(String, Object, int)}
     *
     * @param addr           address addr
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @param timeoutMillis  timeoutMillis timeoutMillis
     * @return object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(final String addr, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeSync(addr, request, requestContext, timeoutMillis);
    }

    /**
     * Synchronous invocation using a parsed {@link Url} <br>
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the parsed {@link Url} to find a available client channel, if none then throw exception</li>
     * </ol>
     *
     * @param url           url
     * @param request       request request
     * @param timeoutMillis timeoutMillis timeoutMillis
     * @return Object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(Url url, Object request, int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeSync(url, request, null, timeoutMillis);
    }

    /**
     * Synchronous invocation with a {@link RequestContext}, common api notice please see {@link #invokeSync(Url, Object, int)}
     *
     * @param url            url
     * @param request        request
     * @param requestContext invokeContext
     * @param timeoutMillis  timeoutMillis
     * @return object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeSync(url, request, requestContext, timeoutMillis);
    }

    /**
     * Synchronous invocation using a {@link Channel} <br>
     * <p>
     * Notice:<br>
     * <b>DO NOT modify the request object concurrently when this method is called.</b>
     *
     * @param channel       channel
     * @param request       request
     * @param timeoutMillis timeoutMillis
     * @return Object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(final Channel channel, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(channel, request, null, timeoutMillis);
    }

    /**
     * Synchronous invocation with a {@link RequestContext}, common api notice please see {@link #invokeSync(Channel, Object, int)}
     *
     * @param channel        channel channel
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @param timeoutMillis  timeoutMillis timeoutMillis
     * @return object
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public Object invokeSync(final Channel channel, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        return this.invoker.invokeSync(channel, request, requestContext, timeoutMillis);
    }

    /**
     * Future invocation using a string address, address format example - 127.0.0.1:12200?key1=value1&key2=value2 <br>
     * You can get result use the returned {@link RpcResponseFuture}.
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the string address to find a available client channel, if none then throw exception</li>
     * <li>Unlike rpc client, address arguments takes no effect here, for rpc server will not create channel.</li>
     * </ol>
     *
     * @param addr          address address
     * @param request       request request
     * @param timeoutMillis timeoutMillis timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public RpcResponseFuture invokeWithFuture(final String addr, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeWithFuture(addr, request, null, timeoutMillis);
    }

    /**
     * Future invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithFuture(String, Object, int)}
     *
     * @param addr           address address
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @param timeoutMillis  timeoutMillis timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public RpcResponseFuture invokeWithFuture(final String addr, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeWithFuture(addr, request, requestContext, timeoutMillis);
    }

    /**
     * Future invocation using a parsed {@link Url} <br>
     * You can get result use the returned {@link RpcResponseFuture}.
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the parsed {@link Url} to find a available client channel, if none then throw exception</li>
     * </ol>
     *
     * @param url           url
     * @param request       request request
     * @param timeoutMillis timeoutMillis timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public RpcResponseFuture invokeWithFuture(final Url url, final Object request, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeWithFuture(url, request, null, timeoutMillis);
    }

    /**
     * Future invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithFuture(Url, Object, int)}
     *
     * @param url            url
     * @param request        request request
     * @param requestContext invokeContext
     * @param timeoutMillis  timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public RpcResponseFuture invokeWithFuture(final Url url, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        return this.invoker.invokeWithFuture(url, request, requestContext, timeoutMillis);
    }

    /**
     * Future invocation using a {@link Channel} <br>
     * You can get result use the returned {@link RpcResponseFuture}.
     * <p>
     * Notice:<br>
     * <b>DO NOT modify the request object concurrently when this method is called.</b>
     *
     * @param channel       channel
     * @param request       request
     * @param timeoutMillis timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException remoting exception
     */
    public RpcResponseFuture invokeWithFuture(final Channel channel, final Object request, final int timeoutMillis) throws RemotingException {

        ensureStarted();
        return this.invoker.invokeWithFuture(channel, request, null, timeoutMillis);
    }

    /**
     * Future invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithFuture(Channel, Object, int)}
     *
     * @param channel        channel channel
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @param timeoutMillis  timeoutMillis timeoutMillis
     * @return RpcResponseFuture
     * @throws RemotingException remoting exception
     */
    public RpcResponseFuture invokeWithFuture(final Channel channel, final Object request, final RequestContext requestContext, final int timeoutMillis) throws RemotingException {

        ensureStarted();
        return this.invoker.invokeWithFuture(channel, request, requestContext, timeoutMillis);
    }

    /**
     * Callback invocation using a string address, address format example - 127.0.0.1:12200?key1=value1&key2=value2 <br>
     * You can specify an implementation of {@link Callback} to get the result.
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the string address to find a available client channel, if none then throw exception</li>
     * <li>Unlike rpc client, address arguments takes no effect here, for rpc server will not create channel.</li>
     * </ol>
     *
     * @param addr          address
     * @param request       request
     * @param callback      invokeCallback
     * @param timeoutMillis timeoutMillis
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void invokeWithCallback(final String addr, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.invokeWithCallback(addr, request, null, callback, timeoutMillis);
    }

    /**
     * Callback invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithCallback(String, Object, Callback, int)}
     *
     * @param addr           address address
     * @param request        request request
     * @param requestContext invokeContext
     * @param callback       invokeCallback
     * @param timeoutMillis  timeoutMillis
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void invokeWithCallback(final String addr, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.invokeWithCallback(addr, request, requestContext, callback, timeoutMillis);
    }

    /**
     * Callback invocation using a parsed {@link Url} <br>
     * You can specify an implementation of {@link Callback} to get the result.
     * <p>
     * Notice:<br>
     * <ol>
     * <li><b>DO NOT modify the request object concurrently when this method is called.</b></li>
     * <li>When do invocation, use the parsed {@link Url} to find a available client channel, if none then throw exception</li>
     * </ol>
     *
     * @param url           url
     * @param request       request request
     * @param callback      invokeCallback invokeCallback
     * @param timeoutMillis timeoutMillis timeoutMillis
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void invokeWithCallback(final Url url, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.invokeWithCallback(url, request, null, callback, timeoutMillis);
    }

    /**
     * Callback invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithCallback(Url, Object, Callback, int)}
     *
     * @param url            url
     * @param request        request request
     * @param requestContext invokeContext invokeContext
     * @param callback       invokeCallback invokeCallback
     * @param timeoutMillis  timeoutMillis timeoutMillis
     * @throws RemotingException    remoting exception remoting exception
     * @throws InterruptedException interrupted exception
     */
    public void invokeWithCallback(final Url url, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException, InterruptedException {
        ensureStarted();
        check();
        this.invoker.invokeWithCallback(url, request, requestContext, callback, timeoutMillis);
    }

    /**
     * Callback invocation using a {@link Channel} <br>
     * You can specify an implementation of {@link Callback} to get the result.
     * <p>
     * Notice:<br>
     * <b>DO NOT modify the request object concurrently when this method is called.</b>
     *
     * @param channel       channel channel
     * @param request       request request
     * @param callback      invokeCallback invokeCallback
     * @param timeoutMillis timeoutMillis  timeoutMillis
     * @throws RemotingException remoting exception remoting exception
     */
    public void invokeWithCallback(final Channel channel, final Object request, final Callback callback, final int timeoutMillis) throws RemotingException {
        ensureStarted();
        this.invoker.invokeWithCallback(channel, request, null, callback, timeoutMillis);
    }

    /**
     * Callback invocation with a {@link RequestContext}, common api notice please see {@link #invokeWithCallback(Channel, Object, Callback, int)}
     *
     * @param channel       channel channel
     * @param request       request request
     * @param callback      invokeCallback invokeCallback
     * @param timeoutMillis timeoutMillis  timeoutMillis
     * @throws RemotingException remoting exception remoting exception
     */
    public void invokeWithCallback(final Channel channel, final Object request, final RequestContext requestContext, final Callback callback, final int timeoutMillis) throws RemotingException {
        ensureStarted();
        this.invoker.invokeWithCallback(channel, request, requestContext, callback, timeoutMillis);
    }

    /**
     * check whether a client address connected
     *
     * @param remoteAddr remote address
     * @return {@code true} means url is connected and {@code false} means disconnected.
     */
    public boolean isConnected(String remoteAddr) {
        ensureStarted();
        Url url = AddressParser.parse(remoteAddr);
        return this.isConnected(url);
    }

    /**
     * check whether a {@link Url} connected
     *
     * @param url url
     * @return {@code true} means url is connected and {@code false} means disconnected.
     */
    public boolean isConnected(Url url) {
        ensureStarted();
        Channel conn = this.invoker.channelManager.get(url.getId());
        if (null != conn) {
            return conn.isFine();
        }
        return false;
    }

    /**
     * check whether channel manage feature enabled
     */
    private void check() {
        if (!option(ServerOption.SERVER_MANAGE_channel_SWITCH)) {
            throw new UnsupportedOperationException("Please enable channel manage feature of Rpc Server before call this method! See comments in constructor RpcServer(int port, boolean managechannel) to find how to enable!");
        }
    }

    /**
     * init netty write buffer water mark
     */
    private void initWriteBufferWaterMark() {
        // init with system properties
        Integer lowWaterMarkConfig = ConfigManager.netty_buffer_low_watermark();
        if (lowWaterMarkConfig != null) {
            option(ServerOption.netty_buffer_low_water_mark, lowWaterMarkConfig);
        }
        Integer highWaterMarkConfig = ConfigManager.netty_buffer_high_watermark();
        if (highWaterMarkConfig != null) {
            option(ServerOption.netty_buffer_high_water_mark, highWaterMarkConfig);
        }

        int lowWaterMark = this.netty_buffer_low_watermark();
        int highWaterMark = this.netty_buffer_high_watermark();
        if (lowWaterMark > highWaterMark) {
            throw new IllegalArgumentException(String.format("[server side]  netty high water mark {%s} should not be smaller than low water mark {%s} bytes)", highWaterMark, lowWaterMark));
        } else {
            //log.warn(
            //    "[server side]  netty low water mark is {} bytes, high water mark is {} bytes",
            //    lowWaterMark, highWaterMark);
        }
        this.serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark, highWaterMark));
    }


    public ChannelManagerImpl getServerChannelManager() {
        return serverChannelManager;
    }
}
